package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * 热门文章流处理
 *
 * @author DiXiaoNie
 * @date 2023/01/07
 * @since 2023/1/7 14:39
 */
@Component
public class HotArticleStreamHandle {

    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //1.接收消息内容
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //2.流式计算
        stream.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            /**
             * 初始化当前接收的消息的
             *
             * @param key   发送消息的key，此时的key为null
             * @param value 发送小的内容，字符串类型，包含的就是
             * @return {@link KeyValue }<{@link String }, {@link String }>
             * @return key=文章id  value=type:add
             * @author DiXiaoNie
             * @date 2023/01/07
             */
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                //1.把value转换成对象
                UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
            }
        })
                //分组
                /*
                 * key=文章id
                 * value = type：add
                 * */
                .groupBy((key, value) -> key)
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                .aggregate(new Initializer<String>() {
                    //初始化
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     * 应用
                     *
                     * @param key   关键
                     * @param value 价值
                     * @param init  初始化
                     * @return {@link String }
                     * @author DiXiaoNie
                     * @date 2023/01/07
                     */
                    @Override
                    public String apply(String key, String value, String init) {
                        //1.判断value是否为空
                        if (value == null) {
                            return init;
                        }
                        //初始化变量
                        Integer collection = 0, comment = 0, like = 0, view = 0;
                        //2.分割value 并且对每种类型的数值进行增量
                        String[] split = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                            case LIKES:
                                like += Integer.parseInt(split[1]);
                                break;
                            case VIEWS:
                                view += Integer.parseInt(split[1]);
                                break;
                            case COMMENT:
                                comment += Integer.parseInt(split[1]);
                                break;
                            case COLLECTION:
                                collection += Integer.parseInt(split[1]);
                                break;
                            default:
                                break;
                        }
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", collection, comment, like, view);

                        return formatStr;
                    }
                    //聚合之后的名称
                }, Materialized.as("myAgg"))
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, String value) {
                        return new KeyValue<>(key.key(), format(key.key(), value));
                    }
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        //3.发送消息内容
        return stream;
    }

    private String format(String key, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(key));
        String[] split = value.split(",");
        for (String agg : split) {
            String[] aggSplit = agg.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(aggSplit[0])) {
                case LIKES:
                    mess.setLike(Integer.parseInt(aggSplit[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(aggSplit[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(aggSplit[1]));
                    break;
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(aggSplit[1]));
                    break;
                default:
                    break;
            }

        }
        return JSON.toJSONString(mess);
    }
}
